package com.innodata.riskEngine.demo.consumer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

/**
 * @Author: liboshuai
 * @Date: 2023-07-28 10:51
 */
@Slf4j
@Component
public class kafkaConsumer {

//    @KafkaListener(topics = {"impRiskWarn"})
//    public void receiveNormal(ConsumerRecord<?,?> consumerRecord) {
//        log.info("topic >>> {}; 分区 >>> {}; 值内容 >>> {}; ", consumerRecord.topic(), consumerRecord.partition(), consumerRecord.value());
//    }
//
//    /**
//     * @Title 指定topic、partition、offset消费
//     * @Description 同时监听topic1和topic2，监听topic1的0号分区、
//     * topic2的 "0号和1号" 分区，指向1号分区的offset初始值为8
//     * @param record
//     */
//    @KafkaListener(id="consumer1",groupId = "felix-group",topicPartitions = {
//            @TopicPartition(topic = "topic1",partitions = {"0"}),
//            @TopicPartition(topic = "topic2",partitions = "0",
//                    partitionOffsets = @PartitionOffset(partition = "1",initialOffset = "8"))
//    })
//    public void onMessage2(ConsumerRecord<?,?> record){
//        System.out.println("topic:"+record.topic()+"partition:"+record.partition()+"offset:"+record.offset()+"value:"+record.value());
//    }
//
//    @KafkaListener(id="consumer2",groupId = "felix-group",topics = "kafka-topic-test-batch" )
//    public void onMesssage(List<ConsumerRecord<?,?>> records){
//        log.info(">>>批量消费一次，records.size()="+records.size());
//        int count = 0;
//        for(ConsumerRecord<?,?> record:records){
//            log.info("批量消费第{}次, 内容为{}", ++count, record.value());
//        }
//    }

}
